package com.pw.study.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

public class $03TableCreate {


    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //初始化表环境
        StreamTableEnvironment sEnv = StreamTableEnvironment.create(env);
        //初始化数据
        Schema schema = new Schema().field("id", DataTypes.STRING())
                .field("ts", DataTypes.BIGINT())
                .field("vc", DataTypes.INT());


        //读文件
        sEnv.connect(new FileSystem().path("data/file/sensor.txt"))
                .withFormat(new Csv().fieldDelimiter(',').lineDelimiter("\n"))
                .withSchema(schema).createTemporaryTable("sensor");
        Table result = sEnv.from("sensor").select($("id"), $("ts"), $("vc"));
        result.printSchema();
        result.execute().print();

        //写入数据
        sEnv.connect(new FileSystem().path("data/file/a.txt")).withFormat(new Csv())
                .withSchema(schema).createTemporaryTable("hello");
        result.executeInsert("hello").print();
        //执行查询
        // mc1(table);
        //mc2(env, sEnv, table);


    }

    private static void mc2(StreamExecutionEnvironment env, StreamTableEnvironment sEnv, Table table) {
        sEnv.toRetractStream(table, Row.class)
                .filter(t -> t.f0).
                map(t -> t.f1).print();
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    private static void mc1(Table table) {
        Table result = table.groupBy($("id"))
                .aggregate($("vc").sum().as("vc_sum"))
                .select($("id"), $("vc_sum"));
        result.printSchema();
        result.execute().print();
    }

}
